package pers.cz.logtail.connect.kafka;


import pers.cz.config.JefConfiguration;
import pers.cz.logtail.connect.Connection;
import pers.cz.logtail.enums.KafkaConfigItem;
import pers.cz.logtail.queue.Message;
import pers.cz.logtail.queue.Producer;
import pers.cz.logtail.queue.QueueConf;
import pers.cz.logtail.queue.kafka.KafkaFactory;

/**
 * @program: postgirl-collection-client
 * @description: Kafka连接实现
 * @author: Cheng Zhi
 * @create: 2022-08-01 17:10
 **/
public class KafkaConnection extends Connection {

    private Producer kafkaProducer;

    public KafkaConnection(QueueConf queueConf) {
        kafkaProducer = KafkaFactory.getInstance(queueConf).getProducer();
    }

    @Override
    public void close() {
        // kafkaProducer.close();
    }

    @Override
    public String get(String key) {
        return null;
    }

    @Override
    public Long autoIncrement(String key) {
        return null;
    }

    @Override
    public void expire(String key, Integer time) {

    }

    @Override
    public void realInsert(String content) {

        String topic = JefConfiguration.get(KafkaConfigItem.KAFKA_TOPIC, "JEF-LOG-TAIL");
        Message message = new Message(topic,content);
        // 异步发送
        kafkaProducer.sendAsync(message);
    }

    @Override
    public void realSet(String key, String value) {

    }

    @Override
    public void realInsert(String key, String context) {

    }

    @Override
    public void realInsert(String ip, String fileName, String content) {
        String defaultKey = ip + "_" + fileName;
        String topic = JefConfiguration.get(KafkaConfigItem.KAFKA_TOPIC, defaultKey);
        Message message = new Message(topic,content);
        // 异步发送
        kafkaProducer.sendAsync(message);
    }
}
